package com.sky.test.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @author: 朱海鹏
 * @Package: com.sky.test.kafka
 * @Project: sky-take-out
 * @date: 2025/2/20 18:53
 * @description :消费者类
 */
public class ConsumerQuickStart {
    /*
    * 注意，在同一个组下的消费者，订阅同一个主题，只能有一个消费者收到消息(一对一)
    *
    * kafka高可用设计
    *    集群
    *   kafka的服务器端由被称为Broker的服务进程构成，即一个Kafka集群由多个Broker组成
    *   这样如果集群中某一台机器宕机，其他机器的Broker也依然能够对外提供服务，这就是Kafka提供高可用的手段之一
    *    备份机制
    *   生成者发送消息时，首先要连上我们的kafka集群，首先连接上的是领导者副本【LEADER REPLICA】,让领导者副本进行数据的存储【就是说领导者副本会自己存一份，
    * 同时为了数据的持久化，也会将数据存储到追随者副本当中【FOLLOWER REPLICA】】
    *    备份机制【Replication】-同步方式
    *     ISR【IN-SYNC-REPLICA】需要同步复制保存的FOLLOWER
    *      如果leader失效后，需要选出新的leader，选举的原则如下：
    *        第一： 选举时优先从ISR中选定，因为这个列表中的FOLLOWER 的数据说与leader同步的
    *        第二：如果ISR列表中的FOLLOWER都不行了，就只能从其他follower中选取
    *     极端情况下，就是所有副本都失效了，这时有两种方案
    *      第一：等待ISR中的第一个活过来，选为Leader，数据可靠，但是活过来的时间不确定
    *      第二:选择第一个活过来的Replication，不一定是ISR的， 选为Leader，以最快的速度恢复可用性，但数据不一定完整
    *   消费者组
    *     消费者组:指的就是由一个或者多个消费者组成的群体
    *     一个发布在Topic上消息被分发在此消息组中的一个消费者
    *        所有的消费者都在一个组上，那么这就变成了queue模型
    *        所有的消费者都在不同的组中，那么就完全变成了发布-订阅模型
    *   消息有序性
    *     topic分区中消息只能由消费者组中唯一一个消费者处理，所以消息肯定是按照先后顺序进行处理的，但是他也仅仅是保证Topic的一个分区顺序处理，不能保证跨区的消息先后处理顺序。所以，如果你想要顺序的处理TOPIC的所有消息
    *     那就只能提供一个分区
    *   提交和偏移量
    *   Kafka不会像其他JMS队列那样要得到消费者的确认，消费者可以使用Kafka来追踪消息在分区的位置【偏移量】
    *   消费者会往一个叫做_consumer_offset的特殊主题发送消息，消息里包含了每个分区的偏移量。如果消费者发生奔溃或有新的消费者加入群组，就会触发再均衡
    *     偏移量
    *       如果提交偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理【重复消费】
    *       如果提交的偏移量大于客户端的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。
    *    偏移量提交方式
    *      提交偏移量的方式有两种，分别是自动提交和手动提交
    *      自动提交
    *         当enable.auto.commit被设置为true，提交方式就是让消费者自动提交偏移量，每隔5秒消费者会自动把从poll()方式接受的最大偏移量提交上去
    *      手动提交
    *         当enable.auto.commit被设置为false可以有以下三种提交方式
    *            提交当前偏移量【同步提交】
    *            异步提交
    *            同步和异步组合提交
    * */

    public static void main(String[] args) {
        // 1: 设置kafka的配置信息
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.200.130:9092");
        // 1.1 key 和 value的反序列化对象
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");

        // 2创建对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 3订阅主题
        consumer.subscribe(Collections.singletonList("topic1"));
        //4 获取信息
        while (true) {
            // 每一秒循环一次
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            records.forEach(record -> {
                System.out.println(record.key());
                System.out.println(record.value());
                // 打印偏移量
                System.out.println(record.offset());
                // 打印当前消息分区号
                System.out.println(record.partition());
            });
        }
    }
}
